package com.atguigu.PracticalProject.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.PracticalProject.app.BaseAppV1;
import com.atguigu.PracticalProject.common.Constant;
import com.atguigu.PracticalProject.util.AtguiguUtil;
import com.atguigu.PracticalProject.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

public class DwdTrafficUniqueVisitorDetail extends BaseAppV1 {
    public static void main(String[] args) {
        new DwdTrafficUniqueVisitorDetail().init(
                3015,
                2,
                "DwdTrafficUniqueVisitorDetail",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        stream
                .map(JSON::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject object, long l) {
                                return object.getLong("ts");
                            }
                        })
                )
                .keyBy( obj -> obj.getJSONObject("common").getString("uid"))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {

                    private ValueState<String> visitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        visitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("visitDateState", String.class));
                    }

                    @Override
                    public void process(String uid,
                                        Context context,
                                        Iterable<JSONObject> iterable,
                                        Collector<JSONObject> collector) throws Exception {
                        //  取出状态
                        String date = visitDateState.value();
                        //  获取当天时间
                        String today = AtguiguUtil.toDate(context.window().getStart());

                        //判断出今天第一个窗口
                        if (!today.equals(date)) {
                            //拿到第一个窗口的所有数据
                            List<JSONObject> list = AtguiguUtil.toList(iterable);
                            //然后比较时间大小，取出时间最小的用户，就是当天第一次登录的用户
                            JSONObject min = Collections.min(list, new Comparator<JSONObject>() {
                                @Override
                                public int compare(JSONObject o1, JSONObject o2) {
                                    return o1.getLong("ts").compareTo(o2.getLong("ts"));
                                }
                            });
                            //把数据抛出
                            collector.collect(min);

                            //更新状态
                            visitDateState.update(today);

                        }
                    }
                })
                .map(obj -> JSON.toJSONString(obj))
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_UV_DETAIL));


    }
}
